package com.atguigu.flink.state;

import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.List;

/**
 * Created by Smexy on 2023/2/2
 *
 *      状态后端:  可拔插组件，负责状态的读写和存储。
 *          hashmap: 默认。
 *                      把状态存储在 TM 的堆内存中。
 *                      存储小状态！
 *
 *
 *          rocksdb: 额外引入。
 *                      flink提供的内嵌式数据库(TM程序启动后，内嵌自带的数据库)，类似hbase。 例如:derby
 *                          列式存储，列族，存储的数据格式都是byte[]。
 *                          数据优先写入到内存中，内存满后自动刷写到磁盘。
 *                      把状态存储在rocksdb数据库中。
 *                      存储大状态，原则上是无限大！
 *
 *                      生产上选择！
 *
 *                      弊端： ①对比hashmap效率低
 *                            ②读或写，都会带来额外的序列化的开销
 *
 *      -------------------------
 *          checkpoint:  负责把状态的数据持久化(备份)到外部设备。
 *
 *              意义: 防止由于进程意外终止，造成状态丢失。
 *
 *              默认： 把状态备份到 JM 的内存中
 *
 *                    生产上选择把状态备份到外部文件系统。
 *
 *
 *
 *
 *
 *
 *
 *
 */
public class Demo8_Checkpoint
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //指定使用Hashmap状态后端，默认
        //env.setStateBackend(new HashMapStateBackend());
        //指定使用rocksdb 状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        env.setParallelism(2);

        //开启ck
        env.enableCheckpointing(2000);
        //把状态备份到本地
        //env.getCheckpointConfig().setCheckpointStorage("file:///D:\\tmp\\ck");
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:9820/ck");

        //设置cancel后依旧保留快照
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        env
            .socketTextStream("hadoop103", 8888)
            .map(new Demo1_State.MyMap())
            .addSink(new SinkFunction<String>()
            {
                @Override
                public void invoke(String value, SinkFunction.Context context) throws Exception {
                    if (value.contains("x")){
                        throw new RuntimeException("出异常了...");
                    }
                    System.out.println(value);
                }
            });



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //累积字符串
    public static class MyMap implements MapFunction<String,String>, CheckpointedFunction
    {

        ListState<String> strState ;

        @Override
        public String map(String value) throws Exception {
            strState.add(value);
            List<String> strs = MyUtil.parseList(strState.get());
            return strs.toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            strState = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("strs",String.class));
        }
    }
}
